package com.second.taobao.util;

import com.typesafe.config.Config;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;

public class HbaseUtils {

  public static final String TABLE_NAME_USER = "topic_user";
  public static final String TABLE_NAME_AD = "topic_ad";
  public static final String TABLE_KEY = "userid";
  private static final String FAMILY_USER = "user";
  private static final String FAMILY_SALE = "sale";
  private static final String FAMILY_BEHAVIOR = "behavior";
  private static final String QUALIFIER_USER_AGE = "age_level";
  private static final String QUALIFIER_USER_OCCUPATION = "occupation";
  private static final String QUALIFIER_USER_CITY = "new_user_class_level";
  private static final String QUALIFIER_SLAE_PVALUE = "pvalue_level";
  private static final String QUALIFIER_SLAE_SHOP = "shopping_level";
  private static final String QUALIFIER_BEHAVIOR_RV_7 = "last_7_day_review";
  private static final String QUALIFIER_BEHAVIOR_RV_14 = "last_14_day_review";
  private static final String QUALIFIER_BEHAVIOR_BUY_7 = "last_7_day_buy";
  private static final String QUALIFIER_BEHAVIOR_BUY_14 = "last_14_day_buy";

  public static Configuration getHBaseConfig() {
    Configuration conf = HBaseConfiguration.create();
    Config hbaseConf = GetConfig.getRoot("hbase");
    conf.set("hbase.zookeeper.quorum", hbaseConf.getString("host"));
    conf.set("zookeeper.znode.parent", hbaseConf.getString("znode.parent"));

    return conf;
  }

  public static Result get(String name, String key) throws IOException {
    Connection connection = null;

    connection = ConnectionFactory.createConnection(getHBaseConfig());

    Table table = connection.getTable(TableName.valueOf(name));
    Get get = new Get(Bytes.toBytes(key));
//    get.addColumn(Bytes.toBytes(FAMILY_USER), Bytes.toBytes(QUALIFIER_USER_AGE));
//    get.addColumn(Bytes.toBytes(FAMILY_SALE), Bytes.toBytes(QUALIFIER_SLAE_PVALUE));
//    get.addColumn(Bytes.toBytes(FAMILY_BEHAVIOR), Bytes.toBytes(QUALIFIER_BEHAVIOR_RV_7));
//    get.addColumn(Bytes.toBytes("article"), Bytes.toBytes("math"));
    Result result = table.get(get);

    table.close();
    connection.close();

    return result;
  }

  public static void put(List<Map<String, String>> list) throws IOException {
    //Connection to the cluster.
    Connection connection = null;
    //a async batch handler
    BufferedMutator bufferedMutator = null;

    //establish the connection to the cluster.
    try {
      connection = ConnectionFactory.createConnection(getHBaseConfig());
      bufferedMutator = connection.getBufferedMutator(TableName.valueOf(TABLE_NAME_USER));
      //describe the data
      for (Map m : list) {
        Put put = new Put(Bytes.toBytes(m.get("userid").toString()));
        put.addColumn(Bytes.toBytes(FAMILY_USER), Bytes.toBytes(QUALIFIER_USER_AGE), Bytes.toBytes(m.get("age").toString()));
        //add data to buffer
        bufferedMutator.mutate(put);
      }
    } catch (IOException e) {
      e.printStackTrace();
    } finally {
      //close
      if (bufferedMutator != null) bufferedMutator.close();
      if (connection != null) connection.close();
    }
  }

}

